package net.bwie.realtime.jtp.dws.douyin.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.dws.douyin.log.bean.*;
import net.bwie.realtime.jtp.dws.douyin.log.functions.DouYinOrderWindowFunction5;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.HashMap;
import java.util.Map;

public class DouYinOrderToTal5 {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink环境（与原类完全一致）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);          // 测试并行度1
        env.enableCheckpointing(3000L); // 3秒Checkpoint

        // 2. 读取两张表的Kafka数据源
        DataStream<String> logKafkaStream = KafkaUtil.consumerKafka(env, "dwd_live_room_event_log");
        DataStream<String> orderKafkaStream = KafkaUtil.consumerKafka(env, "dwd_trade_order_detail");

        // 3. 数据处理：解析→封装载体→合并→窗口聚合→JSON格式化
        DataStream<String> resultStream = handle(logKafkaStream, orderKafkaStream);

        // 4. 打印+输出Doris（与原类逻辑一致）
        resultStream.print("商品统计指标→");
        DorisUtil.saveToDoris(
                resultStream,
                "douyin_realtime_report",
                "dws_goods_stat"
        );

        // 5. 触发作业执行
        env.execute("DouYinOrderToTal5");
    }

    /**
     * 核心处理方法（步骤化，无元组，用UnifiedData承载数据）
     */
    private static DataStream<String> handle(
            DataStream<String> logStream,
            DataStream<String> orderStream) {

        // ------------------------------ 步骤1：解析日志表→封装为UnifiedData ------------------------------
        SingleOutputStreamOperator<UnifiedData5> logUnifiedStream = logStream
                // 解析JSON为EventLog6（与原类一致）
                .map(json -> JSON.parseObject(json, EventLog6.class))
                .filter(event -> event != null) // 过滤空数据
                .filter(event -> "点击".equals(event.getEventType())) // 仅保留点击事件
                // 封装为UnifiedData：flag=1，logData赋值，orderData为null
                .map(event -> {
                    UnifiedData5 data = new UnifiedData5();
                    data.setEventTypeFlag(1); // 1=日志点击事件
                    data.setLogData(event);
                    data.setOrderData(null);
                    return data;
                });

        // ------------------------------ 步骤2：解析订单表→封装为UnifiedData ------------------------------
        SingleOutputStreamOperator<UnifiedData5> orderUnifiedStream = orderStream
                // 解析JSON为OrderDetail6，金额转为double
                .map(json -> {
                    OrderDetail5 order = JSON.parseObject(json, OrderDetail5.class);
                    // 处理金额：JSON字符串→double（避免BigDecimal）
                    String amountStr = JSON.parseObject(json).getString("order_amount");
                    order.setOrderAmount(amountStr != null ? Double.parseDouble(amountStr) : 0.0);
                    return order;
                })
                .filter(order -> order != null) // 过滤空数据
                // 封装为UnifiedData：flag=2，orderData赋值，logData为null
                .map(order -> {
                    UnifiedData5 data = new UnifiedData5();
                    data.setEventTypeFlag(2); // 2=订单成交事件
                    data.setOrderData(order);
                    data.setLogData(null);
                    return data;
                });

        // ------------------------------ 步骤3：合并两张表的UnifiedData流 ------------------------------
        DataStream<UnifiedData5> mergedStream = logUnifiedStream.union(orderUnifiedStream);

        // ------------------------------ 步骤4：按商品ID分组+滑动窗口（与原类一致） ------------------------------
        WindowedStream<UnifiedData5, Long, TimeWindow> windowStream = mergedStream
                // 分组键：商品ID（从UnifiedData中提取）
                .keyBy(unifiedData -> {
                    if (unifiedData.getEventTypeFlag() == 1 && unifiedData.getLogData() != null) {
                        return unifiedData.getLogData().getSkuId(); // 日志表取skuId
                    } else if (unifiedData.getEventTypeFlag() == 2 && unifiedData.getOrderData() != null) {
                        return unifiedData.getOrderData().getSkuId(); // 订单表取skuId
                    }
                    return 0L; // 异常数据归为0组
                })
                // 滑动窗口：5秒窗口+1秒滑动（测试用，生产改5分钟+1分钟）
                .window(SlidingProcessingTimeWindows.of(
                        Time.seconds(5),
                        Time.seconds(1)
                ));

        // ------------------------------ 步骤5：应用窗口函数计算指标 ------------------------------
        SingleOutputStreamOperator<GoodsStatMetric5> metricStream = windowStream
                .apply(new DouYinOrderWindowFunction5()); // 自定义窗口函数

        // ------------------------------ 步骤6：格式化JSON（与原类String.format逻辑一致） ------------------------------
        SingleOutputStreamOperator<String> resultStream = metricStream
                .map(metric -> String.format(
                        "{\"window_start_time\":\"%s\",\"window_end_time\":\"%s\",\"cur_date\":\"%s\",\"sku_id\":\"%d\",\"sku_name\":\"%s\",\"price\":\"%.2f\",\"goods_click_rate\":\"%.4f\",\"deal_order_count\":\"%d\",\"deal_amount\":\"%.2f\",\"goods_conversion_rate\":\"%.4f\"}",
                        // 空值处理（与原类对齐）
                        metric.getWindowStartTime() != null ? metric.getWindowStartTime() : "",
                        metric.getWindowEndTime() != null ? metric.getWindowEndTime() : "",
                        metric.getCurDate() != null ? metric.getCurDate() : "",
                        metric.getSkuId() != null ? metric.getSkuId() : 0,
                        metric.getSkuName() != null ? metric.getSkuName() : "未知商品",
                        metric.getPrice() >= 0 ? metric.getPrice() : 0.00,
                        metric.getGoodsClickRate() >= 0 ? metric.getGoodsClickRate() : 0.0000,
                        metric.getDealOrderCount() != null ? metric.getDealOrderCount() : 0,
                        metric.getDealAmount() >= 0 ? metric.getDealAmount() : 0.00,
                        metric.getGoodsConversionRate() >= 0 ? metric.getGoodsConversionRate() : 0.0000
                ));

        return resultStream;
    }
}

